package com.innodealing.kafka.processor;

import com.innodealing.kafka.annotation.HermesKafkaListener;
import com.innodealing.kafka.exception.UnrecognizedMessageListenerException;
import com.innodealing.kafka.listener.HermesBatchOrderlyListener;
import com.innodealing.kafka.listener.HermesListener;
import com.innodealing.kafka.listener.KafkaBatchMessageListener;
import com.innodealing.kafka.model.MessageListenerType;
import com.innodealing.onshore.datasync.plus.constant.KafkaConstants;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * Hermes kafka 监听器处理器
 *
 * @author Yinpenghai
 */
public class HermesKafkaListenerProcessor implements BeanPostProcessor, EnvironmentAware {
    private final ConsumerFactory<Object, Object> consumerFactory;
    private Environment environment;
    private static final int MAX_CONCURRENCY = 20;

    /**
     * 构造函数
     *
     * @param consumerFactory consumerFactory
     */
    public HermesKafkaListenerProcessor(ConsumerFactory<Object, Object> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) {
        HermesKafkaListener listenerAnnotation = bean.getClass().getAnnotation(HermesKafkaListener.class);
        if (listenerAnnotation != null && bean instanceof HermesListener) {
            String[] topics = listenerAnnotation.topics();
            String[] resolveTopics = resolveTopics(topics);
            String groupId = resolveGroupId(listenerAnnotation.groupId());
            registerListener(bean, resolveTopics, groupId);
        }
        return bean;
    }

    private void registerListener(Object bean, String[] topics, String groupId) {
        if (topics == null || topics.length == 0) {
            throw new IllegalArgumentException("没有指定topic");
        }

        if (Objects.isNull(groupId)) {
            throw new IllegalArgumentException("没有指定groupId");
        }

        ContainerProperties containerProps = new ContainerProperties(topics);
        containerProps.setGroupId(groupId);
        MessageListenerType listenerType = getMessageListenerType(bean);
        if (listenerType == MessageListenerType.BATCH_ORDERLY) {
            containerProps.setAckMode(ContainerProperties.AckMode.MANUAL);
        }

        ConcurrentMessageListenerContainer<Object, Object> container =
                new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);
        final int concurrency = Math.min(topics.length, MAX_CONCURRENCY);
        container.setConcurrency(concurrency);

        if (listenerType.equals(MessageListenerType.BATCH_ORDERLY)) {

            container.setupMessageListener(new KafkaBatchMessageListener((HermesBatchOrderlyListener) bean));
        } else {

            throw new UnrecognizedMessageListenerException("不能识别的MessageListener: " + bean.getClass().getName());
        }
        container.start();
    }

    private MessageListenerType getMessageListenerType(Object bean) {
        if (bean instanceof HermesBatchOrderlyListener) {
            return MessageListenerType.BATCH_ORDERLY;
        } else {
            return MessageListenerType.UNKNOWN;
        }
    }


    private String[] resolveTopics(String[] topics) {
        List<String> result = new ArrayList<>();
        if (topics.length > 0) {
            for (String topic1 : topics) {
                String topic = resolveTopic(topic1);
                result.add(topic);
            }
        }
        return result.toArray(new String[0]);
    }

    private String resolveTopic(String rawTopic) {
        final String resolveTopic = environment.resolvePlaceholders(rawTopic);
        return KafkaConstants.KAFKA_TOPIC_PREFIX + resolveTopic;
    }

    private String resolveGroupId(String groupId) {
        return environment.resolvePlaceholders(groupId);
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }
}
